/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.factory.server.scheduler;

import com.factory.server.algorithms.GraphProblemresult;
import com.factory.server.analyser.Analyser;
import com.factory.server.bo.AbstractBusinessObject;
import com.factory.server.bo.Edge;
import com.factory.server.bo.Fibonacci;
import com.factory.server.bo.Levenshtein;
import com.factory.server.bo.Prime;
import com.factory.server.bo.ResultData;
import com.factory.server.bo.UserInput;
import com.factory.server.bo.User;
import com.factory.server.bo.UserTask;
import com.factory.server.cache.InputCache;
import com.factory.server.cache.OutputCache;
import com.factory.server.consumers.Consumer;
import com.factory.server.consumers.ResultQueue;
import com.factory.server.consumers.SchedulerTasksQueue;
import com.factory.server.consumers.TasksQueue;
import com.factory.server.dao.DAODispatcherIface;
import com.factory.server.dto.EdgeDTO;
import com.factory.server.dto.UserTaskDTO;
import com.factory.server.helper.InputWrapper;
import com.factory.server.helper.MailMail;
import com.factory.server.helper.ProblemEnum;
import com.factory.server.json.SchedulerResult;
import com.factory.server.json.Plan;
import com.factory.server.service.InputQueue;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.Ordering;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

/**
 * Provides implementation for web services published in FactoryREST and other supporting functions.
 * @author milos
 */
@Service
public class SchedulerServiceImpl implements SchedulerServiceIface, ApplicationContextAware {

    static final Logger LOG = Logger.getLogger(SchedulerComputerImpl.class.getName());
    ApplicationContext applicationContext = null;
    private Consumer consumer1 = null;
    private TasksQueue tasksQueue;
    private SchedulerTasksQueue schedulerTasksQueue;
    private ResultQueue resultQueue;
    private InputQueue inputQueue;
    @Autowired
    @Qualifier("daoDispatcher")
    private DAODispatcherIface genericDAODispacher;
    @Autowired
    Analyser analyser;
    @Autowired
    private SchedulerComputerIface schedulerComputer;
    @Autowired
    private MailMail mm;
    private int numberOfThreads = 1;

    public SchedulerServiceImpl() {
        if (tasksQueue == null) {
            tasksQueue = new TasksQueue();
            schedulerTasksQueue = new SchedulerTasksQueue();
        }
    }

    public TasksQueue getTasksQueue() {
        return tasksQueue;
    }

    public SchedulerTasksQueue getSchedulerTaskQueue() {
        return schedulerTasksQueue;
    }

    public void setTasksQueue(TasksQueue tasksQueue) {
        this.tasksQueue = tasksQueue;
    }

    @Override
    public SchedulerResult getSchedulerResult(String inputId) {
        List<UserTask> list = genericDAODispacher.getByProperty("inputId", inputId, UserTask.class);
        UserTask userTask = list.get(0);

        UserTaskDTO userTaskDTO = new UserTaskDTO(userTask);

        return schedulerComputer.computeSchedulerResult(userTaskDTO, schedulerTasksQueue, numberOfThreads);

    }

    public void registerPlan(Plan plan) {
        if (tasksQueue == null) {
            tasksQueue = new TasksQueue();
            schedulerTasksQueue = new SchedulerTasksQueue();
        }
        schedulerTasksQueue.put(plan);
        tasksQueue.put(plan);
        ThreadPoolTaskExecutor taskExecutor = (ThreadPoolTaskExecutor) applicationContext.getBean("taskExecutor");

        if (numberOfThreads == 1) {
            numberOfThreads = taskExecutor.getCorePoolSize() - 2; //one is for input saver and one for result saver
        }

        int i = taskExecutor.getActiveCount();
        consumer1 = (Consumer) applicationContext.getBean("consumer");
        consumer1.setTasksQueue(tasksQueue);
        taskExecutor.execute(consumer1);

        LOG.log(Level.INFO, "Threads count is: " + i);
    }

    public void setApplicationContext(final ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void performPlanTask(Plan plan) {

        String userId = plan.getUserId();
        String problemId = plan.getProblem();
        String inputId = plan.getInputId();

        UserTask userTask = genericDAODispacher.getByProperty("inputId", inputId, UserTask.class).get(0);
        User user = genericDAODispacher.getByProperty("login", userId, User.class).get(0);

        ResultData resultData = new ResultData();
        resultData.setInputId(inputId);
        resultData.setUserId(userId);
        List<AbstractBusinessObject> resultList = new ArrayList<AbstractBusinessObject>();

        if (problemId.equals("SORT")) {
            long computationStartTime = System.currentTimeMillis();
            //Data are in cache ?
            if (InputCache.get(inputId) != null) {
                String inputAsString = InputCache.get(inputId).getDBRecords().get(0).toString();
                resultData.setResult(Joiner.on("\n").join(Ordering.natural().immutableSortedCopy(Splitter.on(" ").split(inputAsString))));

            } else { //data are not in cache
                UserInput searchData = (UserInput) (genericDAODispacher.getByProperty("inputId", inputId, UserInput.class).get(0));
                String inputAsString = searchData.getValue();
                resultData.setResult(Joiner.on("\n").join(Ordering.natural().immutableSortedCopy(Splitter.on(" ").split(inputAsString))));
            }
            LOG.log(Level.INFO, "Pure SORT computation time: " + (System.currentTimeMillis() - computationStartTime));

            LOG.log(Level.INFO, "Storing SORT result started. Time: " + System.currentTimeMillis());

            resultList.add(resultData); //for cache
//            resultQueue.put(resultData);
            this.getResultQueue().put(resultData);

        }

        if (problemId.equals("FIBB")) {
            List<UserInput> indexAsList = genericDAODispacher.getByProperty("inputId", inputId, UserInput.class);
            if (indexAsList.size() == 0) {
                indexAsList = ((List<UserInput>) (List<?>) InputCache.get(inputId).getDBRecords());
            }
            String indexValue = indexAsList.get(0).getValue();
            Fibonacci fibb = analyser.analyseFibb(indexValue);

            resultData.setInputId(inputId);
            resultData.setUserId(userId);
            resultData.setResult(fibb.getValue().toString());

            resultList.add(resultData); //for cache
            resultQueue.put(resultData);
            resultQueue.put(fibb);
        }

        if (problemId.equals("ERATOSTHENES")) {
            List<UserInput> valueAsList = genericDAODispacher.getByProperty("inputId", inputId, UserInput.class);
            if (OutputCache.primes == null) {
                OutputCache.primes = new ArrayList<Prime>();
            }
            Prime maximalPrime = null;
            if (OutputCache.primes.size() > 1) {
                maximalPrime = OutputCache.primes.get(OutputCache.primes.size() - 1);
            }
            if (maximalPrime == null) {
                maximalPrime = new Prime((int) 0);
            }

            if (valueAsList.isEmpty()) {
                valueAsList = ((List<UserInput>) (List<?>) InputCache.get(inputId).getDBRecords());
            }
            String value = valueAsList.get(0).getValue();
            List<Prime> primesAsInteger = analyser.analyseEratosthenes(value, maximalPrime.getValue());

            resultData = new ResultData();
            resultData.setInputId(inputId);
            resultData.setUserId(userId);
            resultData.setResult(value);

            resultList.add(resultData); //for cache
            resultQueue.put(resultData);

            //commented because of memory
            if (primesAsInteger != null) {
                for (Prime prime : primesAsInteger) {
                    if (prime.getValue() > maximalPrime.getValue()) {
                        resultQueue.put(prime);
                    }
                }
            }
            LOG.info("Erathostenes finished.");
        }

        if (problemId.equals("TSP")) {
            List<Edge> edgeList = genericDAODispacher.getByProperty("inputId", inputId, Edge.class);
            List<EdgeDTO> edgeDTOList = new ArrayList<EdgeDTO>();
            Set<Integer> nodesSet = new HashSet<Integer>();
            if (edgeList.isEmpty()) {
                //todo Exception
            }
            for (Edge edge : edgeList) {
                edgeDTOList.add(new EdgeDTO(edge.getSource(), edge.getTarget(), edge.getCost()));
                nodesSet.add(edge.getSource());
                nodesSet.add(edge.getTarget());// TODO treba???
            }
            String TSPResult = analyser.analyseTSP(edgeDTOList);

            //store cost of spanning tree
            resultData.setUserId(userId);
            resultData.setInputId(inputId);
            resultData.setResult("Solution found." + TSPResult);

            resultList.add(resultData); //for cache
            resultQueue.put(resultData);

        }

        if (problemId.equals("SPANN_TREE")) {
            InputWrapper inputFromCache = InputCache.get(inputId);

            List<Edge> edgeList;
            if (inputFromCache != null) {
                edgeList = (List<Edge>) (List<?>) inputFromCache.getDBRecords();
            } else {
                edgeList = genericDAODispacher.getByProperty("inputId", inputId, Edge.class);
            }

            List<EdgeDTO> edgeDTOList = new ArrayList<EdgeDTO>();
            if (edgeList.isEmpty()) {
                edgeList.addAll((List<Edge>) (List<?>) InputCache.get(inputId).getDBRecords());
            }

            HashSet<Integer> nodesSet = new HashSet<Integer>();  //for finding out nodes count

            for (Edge edge : edgeList) {
                EdgeDTO edgeDTO = new EdgeDTO(edge.getSource(), edge.getTarget(), edge.getCost());
                edgeDTOList.add(edgeDTO);
                nodesSet.add(edgeDTO.getSource());
                nodesSet.add(edgeDTO.getTarget());
            }

            GraphProblemresult spanningTreeResult = analyser.analyseSpannTree(edgeDTOList, nodesSet.size(), userTask);

            LOG.log(Level.INFO, "Storing Spanning-Tree result started. Time: " + System.currentTimeMillis());
            for (EdgeDTO edge : spanningTreeResult.getSpanningTree()) { //store edges
                String sourceNode = edge.getSource().toString();
                String targetNode = edge.getTarget().toString();
                double cost = edge.getCost();
                String resultValue = sourceNode + " " + targetNode + " " + cost;

                resultData = new ResultData(userId, inputId, resultValue);
                resultList.add(resultData); //for cache
                resultQueue.put(resultData);
            }
            //store cost of spanning tree
            resultData = new ResultData(userId, inputId, Double.toString(spanningTreeResult.getCost()));
            resultList.add(resultData); //for cache
            resultQueue.put(resultData);
        }

        if (problemId.equals("SHORTEST_PATH")) {
            InputWrapper inputFromCache = InputCache.get(inputId);
            List<Edge> edgeList;
            if (inputFromCache != null) {
                edgeList = (List<Edge>) (List<?>) inputFromCache.getDBRecords();
            } else {
                edgeList = genericDAODispacher.getByProperty("inputId", inputId, Edge.class);
            }
            List<EdgeDTO> edgeDTOList = new ArrayList<EdgeDTO>();
            if (edgeList.isEmpty()) {
                edgeList.addAll((List<Edge>) (List<?>) InputCache.get(inputId).getDBRecords());
            }
            GraphProblemresult SPTResult = analyser.analyseSPT(edgeList, userTask);
            Integer sourceNode;
            Integer targetNode;
            String resultValue;
            LOG.log(Level.INFO, "Storing SPT result started. Time: " + System.currentTimeMillis());
            boolean previousWasDelimiter = false;
            for (EdgeDTO edgeDTO : SPTResult.getShortestPath()) { //store edges
                if (edgeDTO.getSource() == null) {  //empty edge as new line in result file
                    if (!previousWasDelimiter) {
                        resultValue = "*******";
                        previousWasDelimiter = true;
                    } else {
                        continue;
                    }
                } else {
                    sourceNode = edgeDTO.getSource();
                    targetNode = edgeDTO.getTarget();
                    double cost = edgeDTO.getCost();
                    resultValue = sourceNode + " " + targetNode + " " + Double.toString(cost);
                    previousWasDelimiter = false;
                }

                resultData = new ResultData(userId, inputId, resultValue);
                resultList.add(resultData); //for cache
                resultQueue.put(resultData);
            }
            //store cost of shortest path
//            if (SPTResult.getCost() != 0) {
//                resultList.add(resultData); //for cache
//                resultQueue.put(resultData);
//            }
        }

        if (problemId.equals("TOPOLOGICAL_ORDER")) {

            List<Edge> edgeList = new ArrayList<Edge>();
            if (InputCache.get(inputId) != null) {
                edgeList.addAll((List<Edge>) (List<?>) InputCache.get(inputId).getDBRecords());
            }
            List<EdgeDTO> edgeDTOList = new ArrayList<EdgeDTO>();
            if (edgeList.isEmpty()) {
                edgeList = genericDAODispacher.getByProperty("inputId", inputId, Edge.class);
            }
            Integer numberOfNodes = Integer.parseInt(userTask.getNote());
            com.factory.server.algorithms.Graph graph = new com.factory.server.algorithms.Graph(numberOfNodes);

            for (Edge edge : edgeList) {
                edgeDTOList.add(new EdgeDTO(edge.getSource(), edge.getTarget(), edge.getCost()));
                graph.addEdge(edge.getSource(), edge.getTarget());
            }
            GraphProblemresult topologicalOrderResult = analyser.analyseTopologicalOrder(graph);
            List<Integer> topologicalOrder = topologicalOrderResult.getNodesList();
            String resultValue = String.valueOf(topologicalOrder.remove(0));
            for (Integer node : topologicalOrder) { //store edges
                resultValue = resultValue + " -> " + node;
            }
            resultData.setResult(resultValue);

            resultList.add(resultData); //for cache
            resultQueue.put(resultData);

        }

        if (problemId.equals(
                "LEVENSHTEIN")) {
            List<UserInput> textAsList = genericDAODispacher.getByProperty("inputId", inputId, UserInput.class);
            String text = Joiner.on(" ").join(textAsList);

            List<Levenshtein> levenshteinBOList = genericDAODispacher.getByProperty("inputId", inputId, Levenshtein.class);
            if (levenshteinBOList == null || levenshteinBOList.isEmpty()) {
                levenshteinBOList.addAll((List<Levenshtein>) (List<?>) InputCache.get(inputId).getDBRecords());
            }
            String pattern = levenshteinBOList.get(0).getPattern();
            int distance = levenshteinBOList.get(0).getDistance();

            List<Integer> result = analyser.analyseLevenshtein(text, pattern, distance);

            LOG.log(Level.INFO, "Storing Levenstein result started. Time: " + System.currentTimeMillis());

            if (result.isEmpty()) {
                resultData = new ResultData();
                resultData.setInputId(inputId);
                resultData.setUserId(userId);
                resultData.setResult("none");
                resultList.add(resultData); //for cache
            } else {
                for (Integer i : result) {
                    resultData = new ResultData();
                    resultData.setInputId(inputId);
                    resultData.setUserId(userId);
                    resultData.setResult(i.toString());
                    resultList.add(resultData); //for cache
                }
            }
            resultQueue.put(resultData);
        }
        //put result to cache
        OutputCache.add(inputId, resultList);

        //change flag in UserTask that task is finished
        UserTask task = genericDAODispacher.getByProperty("inputId", inputId, UserTask.class).get(0);
        task.setDone(true);
        genericDAODispacher.saveOrUpdate(task);
        try {
            mm.sendMail("miloss.kovalcik@gmail.com",
                    user.getEmail(),
                    "Problem resolved.",
                    String.format("Your problem %s has been resolved, and result is ready to download.", ProblemEnum.valueOf(problemId).getCode()));
        } catch (Exception ex) {
            LOG.info("Sending notification mail failed.");
        }

    }

    public int getNumberOfThreads() {
        return numberOfThreads;
    }

    public void setNumberOfThreads(int numberOfThreads) {
        this.numberOfThreads = numberOfThreads;
    }

    public ResultQueue getResultQueue() {
        return resultQueue;
    }

    public InputQueue getInputQueue() {
        return inputQueue;
    }

    public void setResultQueue(ResultQueue resultQueue) {
        this.resultQueue = resultQueue;
    }

    public void setInputQueue(InputQueue inputQueue) {
        this.inputQueue = inputQueue;
    }
}
